﻿#include"threadpool.h"
#include<pthread.h>

extern const int NUMBER = 2;//增加线程个数
const double ratio = 1.5;//缩减线程，触发条件（存活线程/在忙线程）

//线程池创建函数
ThreadPool* threadPoolCreate(int min, int max, int maxqueueSize)//最小线程数，最大线程数，任务队列容量
{
	ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));//创建线程池

	do {
		if (pool == NULL)//线程池创建失败
		{
			printf("malloc threadpool fail...\n");
			break;
		}
		pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * max);//创建线程队列
		//线程队列创建失败
		if (pool->threadIDs == NULL)
		{
			printf("malloc threadIDs fail...\n");
			break;
		}
		memset(pool->threadIDs, 0, sizeof(pthread_t) * max);//设置默认值
		pool->minNum = min;
		pool->maxNum = max;
		pool->busyNum = 0;		//在忙线程数
		pool->liveNum = min;    //存活线程数，初始最小线程数数相等
		pool->exitNum = 0;		//将要销毁线程

		//互斥锁，条件变量 初始化
		if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 ||
			pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
			pthread_cond_init(&pool->notEmpty, NULL) != 0 ||
			pthread_cond_init(&pool->notFull, NULL) != 0)
		{
			printf("mutex or condition init fail...\n");
			break;
		}

		// 任务队列初始化
		pool->taskQ = (Task*)malloc(sizeof(Task) * maxqueueSize);
		pool->queueCapacity = maxqueueSize;
		pool->queueSize = 0;//队长
		pool->queueFront = 0;//队头
		pool->queueRear = 0;//队尾

		pool->shutdown = 0;//开关状态标记

		// 创建管理者线程，绑定任务，线程池为任务参数
		pthread_create(&pool->managerID, NULL, manager_1, pool);
		for (int i = 0; i < min; ++i)//创建工作线程
		{
			pthread_create(&pool->threadIDs[i], NULL, worker, pool);
		}

		return pool;
	} while (0);// 使用do while 确保所有创建任务都能一次性完成

	//否则,释放所有可能已经创建的资源
	if (pool && pool->threadIDs) free(pool->threadIDs);
	if (pool && pool->taskQ) free(pool->taskQ);
	if (pool) free(pool);

	return NULL;
}

//线程池销毁
int threadPoolDestroy(ThreadPool* pool)//销毁线程池
{
	if (pool == NULL)
	{
		return -1;
	}

	pool->shutdown = 1;// 标记线程池状态，通知其他线程，线程池已关闭

	pthread_join(pool->managerID, NULL);	// 阻塞回收，管理者线程

	// 唤醒阻塞的消费者线程（无任务，休眠的线程）
	for (int i = 0; i < pool->liveNum; ++i)
	{
		pthread_cond_signal(&pool->notEmpty);
	}
	// 释放堆内存
	if (pool->taskQ)
	{
		free(pool->taskQ);
	}
	if (pool->threadIDs)
	{
		free(pool->threadIDs);
	}
	//销毁互斥锁和条件变量
	pthread_mutex_destroy(&pool->mutexPool);
	pthread_mutex_destroy(&pool->mutexBusy);
	pthread_cond_destroy(&pool->notEmpty);
	pthread_cond_destroy(&pool->notFull);

	free(pool);
	pool = NULL;

	return 0;
}

//向任务队列添加任务
void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg)
{
	//不能添加任务的情况
	pthread_mutex_lock(&pool->mutexPool);//锁住线程池互斥锁
	//任务队列满，线程池未关闭
	while (pool->queueSize == pool->queueCapacity && !pool->shutdown)
	{
		pthread_cond_wait(&pool->notFull, &pool->mutexPool);// 阻塞生产者线程，释放互斥锁
	}
	//线程池已经关闭
	if (pool->shutdown)
	{
		pthread_mutex_unlock(&pool->mutexPool);//释放线程池互斥锁
		return;
	}
	//添加任务到任务队列
	pool->taskQ[pool->queueRear].function = func;
	pool->taskQ[pool->queueRear].arg = arg;
	pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity;//循环队列
	pool->queueSize++;

	pthread_cond_signal(&pool->notEmpty);//通知一个线程有任务了
	pthread_mutex_unlock(&pool->mutexPool);//释放线程池互斥锁
}

//获取在忙线程数
int threadPoolBusyNum(ThreadPool* pool)
{
	pthread_mutex_lock(&pool->mutexBusy);//锁住在忙线程数量，（可用读写锁实现，又因为管理者不需要多线程并发，故使用普通锁）
	int busyNum = pool->busyNum;
	pthread_mutex_unlock(&pool->mutexBusy);//解锁
	return busyNum;
}

//获取存活线程数
int threadPoolAliveNum(ThreadPool* pool)
{
	pthread_mutex_lock(&pool->mutexPool);//同threadPoolBusyNum
	int aliveNum = pool->liveNum;
	pthread_mutex_unlock(&pool->mutexPool);
	return aliveNum;
}

//退出当前线程
void threadExit(ThreadPool* pool)
{
	pthread_t tid = pthread_self();//获取当前线程ID
	//在线程队列中查找当前线程，标记退出
	for (int i = 0; i < pool->maxNum; ++i)
	{
		if (pool->threadIDs[i] == tid)
		{
			pool->threadIDs[i] = 0;
			printf("threadExit() called, %ld exiting...\n", tid);
			break;
		}
	}
	pthread_exit(NULL);//当前线程退出
}

//工作的线程(消费者线程)任务函数
void* worker(void* arg)
{
	ThreadPool* pool = (ThreadPool*)arg;//获取线程池“权限”

	//使用while true 实现任务完成就去抢任务
	while (1)
	{
		if (pool->shutdown == 1)	break;
		pthread_mutex_lock(&pool->mutexPool);//锁住线程池
//任务队列空
		// 当前任务队列为空，线程池未关闭

		/*
			此处使用while循环，在某次条件变量等待被解除后，再次检查任务队列和线程池的状态
			why? 因为条件变量也通知了其他线程，任务可能被抢先领走，实际上并没有任务可认领，这便是————“虚假唤醒”问题
			所以，要重新检查条件
			如果使用if 则只检查一次，当线程被阻塞，而后被条件变量唤醒，“线程就会直接继续运行之后的操作”，没有任务却在工作，产生错误！
		*/
		while (pool->queueSize == 0 && !pool->shutdown)
		{
			pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);// 没有任务 阻塞工作线程，释放线程池锁，等待任务通知

			// 线程唤醒后，*重新锁住线程池* 判断自己是不是多余，是不是要销毁线程，执行销毁操作 见 manager()
			if (pool->exitNum > 0)
			{
				pool->exitNum--;
				//存活数大于最小值
				if (pool->liveNum > pool->minNum)
				{
					pool->liveNum--;
					pthread_mutex_unlock(&pool->mutexPool);//释放线程池锁
					threadExit(pool);//当前线程退出，“自毁”，我杀我自己
				}
			}
		}
		// 线程池被关闭了，家没了
		if (pool->shutdown)
		{
			pthread_mutex_unlock(&pool->mutexPool);
			threadExit(pool);//自毁
		}
		//任务队列不为空，		读到这里先别晕，抬头看看线程池还是锁着的
				// 从任务队列中取出一个任务
		Task task;
		task.function = pool->taskQ[pool->queueFront].function;
		task.arg = pool->taskQ[pool->queueFront].arg;

		// 移动头结点，任务个数减1
		pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity;
		pool->queueSize--;

		pthread_cond_signal(&pool->notFull);//加入任务，通知阻塞在条件变量notfull上一个的线程解除阻塞

		pthread_mutex_unlock(&pool->mutexPool);// 解锁线程池

		printf("thread %ld start working...\n", pthread_self());

		pthread_mutex_lock(&pool->mutexBusy);//当前线程获取任务，在忙，busyNum++
		pool->busyNum++;
		pthread_mutex_unlock(&pool->mutexBusy);

		task.function(task.arg);//处理任务
		free(task.arg);//*任务完成，清理任务，与测试函数相关*
		task.arg = NULL;

		printf("thread %ld finish one work\n", pthread_self());

		pthread_mutex_lock(&pool->mutexBusy);//在忙数减1
		pool->busyNum--;
		pthread_mutex_unlock(&pool->mutexBusy);
	}
	return NULL;
}
//管理线程池，增加线程，减少线程（火车司机，烧大火，烧小火，但不能砸火车）
void* manager(void* arg)
{
	ThreadPool* pool = (ThreadPool*)arg;//获得线程池权限
	//线程池没有关闭
	while (!pool->shutdown)
	{
		// 每隔3s检测一次
		sleep(3);//linux 系统特有，与windows不同

		pthread_mutex_lock(&pool->mutexPool);// 取出线程池中 任务的数量 和 当前线程 的数量
		int queueSize = pool->queueSize;
		int liveNum = pool->liveNum;
		pthread_mutex_unlock(&pool->mutexPool);

		pthread_mutex_lock(&pool->mutexBusy); // 取出忙的线程的数量
		int busyNum = pool->busyNum;
		pthread_mutex_unlock(&pool->mutexBusy);

		// 添加线程
			// 任务的个数>存活的线程个数 && 存活的线程数<最大线程数
		if (queueSize > liveNum && liveNum < pool->maxNum)
		{
			pthread_mutex_lock(&pool->mutexPool);
			int counter = 0;
			for (int i = 0; i < pool->maxNum && counter < NUMBER
				&& pool->liveNum < pool->maxNum; ++i)
			{
				if (pool->threadIDs[i] == 0)//查找线程容器中可加入线程的空位
				{
					pthread_create(&pool->threadIDs[i], NULL, worker, pool);
					counter++;
					pool->liveNum++;
				}
			}
			pthread_mutex_unlock(&pool->mutexPool);
		}
		// 销毁线程
			// 忙的线程*ratio < 存活的线程数 && 存活的线程>最小线程数
		if (busyNum * ratio < liveNum && liveNum > pool->minNum)
		{
			pthread_mutex_lock(&pool->mutexPool);
			pool->exitNum = NUMBER;
			pthread_mutex_unlock(&pool->mutexPool);
			//让在 任务队列为空&&线程池没有关闭 条件下 阻塞在 notEmpty上 的 worker 检查exitNum，自毁
			for (int i = 0; i < NUMBER; ++i)
			{
				pthread_cond_signal(&pool->notEmpty);//通知一个阻塞的线程
			}
		}
	}
	return NULL;
}

void* manager_1(void* arg)
{
	ThreadPool* pool = (ThreadPool*)arg;//获得线程池权限
	//线程池没有关闭
	while (!pool->shutdown)
	{
		// 每隔3s检测一次
		sleep(3);

		pthread_mutex_lock(&pool->mutexPool);// 取出线程池中 任务的数量 和 当前线程 的数量
		int queueSize = pool->queueSize;
		int liveNum = pool->liveNum;
		int busyNum = pool->busyNum; // 取出忙的线程的数量

// 添加线程
		// 任务的个数>存活的线程个数 && 存活的线程数<最大线程数
		if (queueSize > liveNum && liveNum < pool->maxNum)
		{
			int counter = 0;
			for (int i = 0; i < pool->maxNum && counter < NUMBER
				&& pool->liveNum < pool->maxNum; ++i)
			{
				if (pool->threadIDs[i] == 0)
				{
					pthread_create(&pool->threadIDs[i], NULL, worker, pool);
					counter++;
					pool->liveNum++;
				}
			}
		}
		// 销毁线程
			// 忙的线程*ratio 比例 < 存活的线程数 && 存活的线程>最小线程数
		if (busyNum * ratio < liveNum && liveNum > pool->minNum)
		{
			pool->exitNum = NUMBER;	//退出线程数量有待商榷
			// 让工作的线程自杀
			for (int i = 0; i < NUMBER; ++i)
			{
				pthread_cond_signal(&pool->notEmpty);
			}
		}
		pthread_mutex_unlock(&pool->mutexPool);
	}

	return NULL;
}